<?php

namespace core;

use Exception;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

class CRabbitMQSimple extends CSingleton
{
    protected ?AMQPStreamConnection $connection = null;
    protected ?AMQPChannel $channel = null;
    protected string $queue = 'q.normal.test';
    protected bool $durable = true;
    protected bool $no_ack = false;
    private bool $publisher_confirms = false;

    public function connect($host = 'localhost', $port = 5672, $user = 'li', $password = 'ioio', $vhost = "/"): static
    {
        // create a connection to the server
        $this->connection = new AMQPStreamConnection($host, $port, $user, $password, $vhost);
        //  create a channel, which is where most of the API for getting things done resides
        $this->channel = $this->connection->channel();

        if ($this->publisher_confirms) $this->channel->confirm_select();

        return $this;
    }

    public function consume($sleep = 0)
    {
        $callback = function (AMQPMessage $msg) use ($sleep) {
            $date = date("Y-m-d H:i:s");
            echo " [x] $date Received ", $msg->body, "\n";
            if ($sleep) sleep($sleep);
            echo " [x] Done\n";
            // If a consumer dies (its channel is closed, connection is closed, or TCP connection is lost) without sending an ack, RabbitMQ will understand that a message wasn't processed fully and will re-queue it.
            // If there are other consumers online at the same time, it will then quickly redeliver it to another consumer.
            $msg->ack();
        };
        // This tells RabbitMQ not to give more than one message to a worker at a time.
        $this->channel->basic_qos(null, 1, null);
        $this->channel->basic_consume($this->queue, '', false, $this->no_ack, false, false, $callback);

        while ($this->channel->is_open()) {

            $this->channel->wait();
        }
    }

    public function setQueue(string $queue = ''): static
    {
        $this->queue = $queue;

        return $this;
    }

    /**
     * @param bool $confirm
     * @return $this
     */
    public function setPublisherConfirms(bool $confirm = true): static
    {
        $this->publisher_confirms = $confirm;

        return $this;
    }

    /**
     * @param bool $no_ack
     * @return $this
     */
    public function setNoAck(bool $no_ack = true): static
    {
        $this->no_ack = $no_ack;

        return $this;
    }

    /**
     * @param bool $durable 持久化
     * @return $this
     */
    public function setDurable(bool $durable = true): static
    {
        $this->durable = $durable;

        return $this;
    }

    public function send($message)
    {
        // declare the queue from which we're going to consume
        $this->channel->queue_declare($this->queue, false, $this->durable, false, false);

        $properties = [
            'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
        ];

        $msg = new AMQPMessage($message, $properties);

        $this->channel->basic_publish($msg, '', $this->queue);

        if ($this->publisher_confirms) $this->channel->wait_for_pending_acks(2.16);
    }



    public function __destruct()
    {
        try {
            // Lastly, we close the channel and the connection
            $this->channel?->close();
            $this->connection?->close();
        } catch (Exception $e) {
            CLog::error(__FILE__, ['error' => $e->getMessage()]);
        }
    }


}